package com.atguigu.tingshu.album.service.impl;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.lang.Assert;
import cn.hutool.core.util.IdUtil;
import com.atguigu.tingshu.album.mapper.*;
import com.atguigu.tingshu.album.service.AlbumInfoService;
import com.atguigu.tingshu.common.cache.GuiGuCache;
import com.atguigu.tingshu.common.constant.KafkaConstant;
import com.atguigu.tingshu.common.constant.RedisConstant;
import com.atguigu.tingshu.common.constant.SystemConstant;
import com.atguigu.tingshu.common.execption.GuiguException;
import com.atguigu.tingshu.common.result.ResultCodeEnum;
import com.atguigu.tingshu.common.service.KafkaService;
import com.atguigu.tingshu.model.album.AlbumAttributeValue;
import com.atguigu.tingshu.model.album.AlbumInfo;
import com.atguigu.tingshu.model.album.AlbumStat;
import com.atguigu.tingshu.model.album.TrackInfo;
import com.atguigu.tingshu.model.search.AlbumInfoIndex;
import com.atguigu.tingshu.query.album.AlbumInfoQuery;
import com.atguigu.tingshu.user.client.UserFeignClient;
import com.atguigu.tingshu.vo.album.*;
import com.atguigu.tingshu.vo.user.UserInfoVo;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cache.annotation.Cacheable;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.Arrays;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@Slf4j
@Service
@SuppressWarnings({"all"})
public class AlbumInfoServiceImpl extends ServiceImpl<AlbumInfoMapper, AlbumInfo> implements AlbumInfoService {

    @Autowired
    private AlbumInfoMapper albumInfoMapper;

    @Autowired
    private AlbumAttributeValueMapper albumAttributeValueMapper;

    @Autowired
    private AlbumStatMapper albumStatMapper;

    @Autowired
    private TrackInfoMapper trackInfoMapper;

    @Autowired
    private KafkaService kafkaService;

    @Autowired
    private UserFeignClient userFeignClient;

    @Autowired
    private TrackStatMapper trackStatMapper;

    /**
     * 保存专辑方法
     * 1.将提交专辑VO转为PO对象，新增专辑
     * 2.将提交专辑标签封装为专辑标签关系集合对象，进行批量保存
     * 3.为新增专辑，批量保存专辑统计信息
     *
     * @param userId      用户ID
     * @param albumInfoVo 专辑信息VO对象
     */
    @Override
    @Transactional(rollbackFor = Exception.class) //默认事务注解修饰方法捕获运行时异常或者Error
    public void saveAlbumInfo(Long userId, AlbumInfoVo albumInfoVo) {
        //1.保存专辑
        //1.1 将提交专辑VO拷贝到专辑PO对象中 仅限同名同类型属性才会被拷贝
        AlbumInfo albumInfo = BeanUtil.copyProperties(albumInfoVo, AlbumInfo.class);

        //1.2 手动为用户id、包含声音数量（0）、是否完结、免费试听集数、审核状态（TODO：审核通过，实则有内容审核机制）
        albumInfo.setUserId(userId);
        albumInfo.setIncludeTrackCount(0);
        albumInfo.setIsFinished("0");
        albumInfo.setTracksForFree(5);
        albumInfo.setStatus(SystemConstant.ALBUM_STATUS_PASS);
        //1.3 保存专辑，得到专辑ID
        albumInfoMapper.insert(albumInfo);
        Long albumId = albumInfo.getId();

        //2.保存专辑标签关系
        //2.1 将提交专辑标签关系VO集合转为PO集合
        List<AlbumAttributeValueVo> albumAttributeValueVoList = albumInfoVo.getAlbumAttributeValueVoList();
        if (CollectionUtil.isNotEmpty(albumAttributeValueVoList)) {
            //2.2 关联专辑ID批量保存专辑标签关系
            for (AlbumAttributeValueVo albumAttributeValueVo : albumAttributeValueVoList) {
                AlbumAttributeValue albumAttributeValue = BeanUtil.copyProperties(albumAttributeValueVo, AlbumAttributeValue.class);
                albumAttributeValue.setAlbumId(albumId);
                albumAttributeValueMapper.insert(albumAttributeValue);
            }

        }
        //3.保存专辑统计信息（4条）
        this.saveAlbumStat(albumId, SystemConstant.ALBUM_STAT_PLAY, 0);
        this.saveAlbumStat(albumId, SystemConstant.ALBUM_STAT_SUBSCRIBE, 0);
        this.saveAlbumStat(albumId, SystemConstant.ALBUM_STAT_BUY, 0);
        this.saveAlbumStat(albumId, SystemConstant.ALBUM_STAT_COMMENT, 0);

        //4.TODO 调用内容审核接口（第三方阿里云）对专辑内容（封面、文字）

        //5.审核通过后发送上架专辑消息到Kafka
        if (true && "1".equals(albumInfo.getIsOpen())) {
            kafkaService.sendMessage(KafkaConstant.QUEUE_ALBUM_UPPER, albumId.toString());
        }
    }

    /**
     * 保存专辑统计信息
     *
     * @param albumId  专辑ID
     * @param statType 统计类型
     * @param statNum  统计数量
     */
    @Override
    public void saveAlbumStat(Long albumId, String statType, int statNum) {
        AlbumStat albumStat = new AlbumStat();
        albumStat.setAlbumId(albumId);
        albumStat.setStatType(statType);
        albumStat.setStatNum(statNum);
        albumStatMapper.insert(albumStat);
    }

    /**
     * 查询当前用户专辑分页列表
     *
     * @param pageInfo       分页对象
     * @param albumInfoQuery 查询条件对象
     * @return
     */
    @Override
    public Page<AlbumListVo> getUserAlbumPage(Page<AlbumListVo> pageInfo, AlbumInfoQuery albumInfoQuery) {
        return albumInfoMapper.getUserAlbumPage(pageInfo, albumInfoQuery);
    }


    /**
     * 根据专辑ID删除专辑
     *
     * @param id
     * @return
     */
    @Override
    @Transactional(rollbackFor = Exception.class)
    public void removeAlbumInfo(Long id) {
        //1.根据专辑ID查询该专辑下包含声音 如果关联声音则不允许删除
        LambdaQueryWrapper<TrackInfo> queryWrapper = new LambdaQueryWrapper<>();
        queryWrapper.eq(TrackInfo::getAlbumId, id);
        Long count = trackInfoMapper.selectCount(queryWrapper);
        if (count > 0) {
            throw new GuiguException(400, "该专辑下存在未删除声音！");
        }
        //2.根据主键删除专辑
        albumInfoMapper.deleteById(id);

        //3.根据专辑ID删除专辑标签
        LambdaQueryWrapper<AlbumAttributeValue> albumAttributeValueLambdaQueryWrapper = new LambdaQueryWrapper<>();
        albumAttributeValueLambdaQueryWrapper.eq(AlbumAttributeValue::getAlbumId, id);
        albumAttributeValueMapper.delete(albumAttributeValueLambdaQueryWrapper);

        //4.根据专辑ID删除专辑统计数据
        LambdaQueryWrapper<AlbumStat> albumStatLambdaQueryWrapper = new LambdaQueryWrapper<>();
        albumStatLambdaQueryWrapper.eq(AlbumStat::getAlbumId, id);
        albumStatMapper.delete(albumStatLambdaQueryWrapper);

        //5.发送Kafka消息通知搜索服务专辑下架
        kafkaService.sendMessage(KafkaConstant.QUEUE_ALBUM_LOWER, id.toString());
    }

    @Autowired
    private RedissonClient redissonClient;

    /**
     * 获取专辑信息（Redissson实现分布式锁）避免缓存击穿
     *
     * @param id
     * @return
     */
    @Override
    public AlbumInfo getAlbumInfo(Long id) {
        try {
            //1.优先从缓存中获取数据
            //1.1 构建业务数据Key 形式：前缀+业务唯一标识
            String dataKey = RedisConstant.ALBUM_INFO_PREFIX + id;
            //1.2 查询Redis获取业务数据
            AlbumInfo albumInfo = (AlbumInfo) redisTemplate.opsForValue().get(dataKey);
            //1.3 命中缓存则直接返回
            if (albumInfo != null) {
                log.info("命中缓存，直接返回，线程ID：{}，线程名称：{}", Thread.currentThread().getId(), Thread.currentThread().getName());
                return albumInfo;
            }
            //2.获取分布式锁
            //2.1 构建锁key
            String lockKey = RedisConstant.ALBUM_INFO_PREFIX + id + RedisConstant.CACHE_LOCK_SUFFIX;
            //2.1 基于RedissonClient创建锁对象 入参成为Redis中Hash结构锁的key
            RLock lock = redissonClient.getLock(lockKey);
            //2.2 获取锁-阻塞直到获取锁成功-lock方法 入参释放锁时间-1（开启看门狗） 默认锁时间：30s
            lock.lock();

            try {
                //3.获取锁成功执行业务,将查询业务数据放入缓存Redis
                //3.1 处于阻塞等待获取线程（终将获取锁成功）避免获取锁线程再次查库，这里再查一次缓存
                albumInfo = (AlbumInfo) redisTemplate.opsForValue().get(dataKey);
                if (albumInfo != null) {
                    log.info("当前线程{},获取锁成功，且再次命中缓存成功", Thread.currentThread().getName());
                    return albumInfo;
                }
                //3.2 查询数据库将查询结果放入缓存Redis 返回
                albumInfo = this.getAlbumInfoFromDB(id);
                long ttl = albumInfo == null ? RedisConstant.ALBUM_TEMPORARY_TIMEOUT : RedisConstant.ALBUM_TIMEOUT;
                redisTemplate.opsForValue().set(dataKey, albumInfo, ttl, TimeUnit.SECONDS);
                return albumInfo;
            } finally {
                //4.业务执行完毕释放锁
                log.info("当前线程：{}，释放锁", Thread.currentThread().getName());
                lock.unlock();
            }
        } catch (Exception e) {
            //5.兜底处理方案：Redis服务有问题，将业务数据获取自动从数据库获取
            log.error("[专辑服务]查询专辑数据异常：{}", e);
            return this.getAlbumInfoFromDB(id);
        }
    }
    /**
     * 获取专辑信息（SpringDataRedis实现分布式锁）避免缓存击穿
     *
     * @param id
     * @return
    @Override
    public AlbumInfo getAlbumInfo(Long id) {
        try {
            //1.优先从缓存中获取数据
            //1.1 构建业务数据Key 形式：前缀+业务唯一标识
            String dataKey = RedisConstant.ALBUM_INFO_PREFIX + id;
            //1.2 查询Redis获取业务数据
            AlbumInfo albumInfo = (AlbumInfo) redisTemplate.opsForValue().get(dataKey);
            //1.3 命中缓存则直接返回
            if (albumInfo != null) {
                log.info("命中缓存，直接返回，线程ID：{}，线程名称：{}", Thread.currentThread().getId(), Thread.currentThread().getName());
                return albumInfo;
            }
            //2.尝试获取分布式锁（set k v ex nx可能获取锁失败）
            //2.1 构建锁key
            String lockKey = RedisConstant.ALBUM_INFO_PREFIX + id + RedisConstant.CACHE_LOCK_SUFFIX;
            //2.2 采用UUID作为线程标识
            String lockVal = IdUtil.fastSimpleUUID();
            //2.3 利用Redis提供set nx ex 获取分布式锁
            Boolean flag = redisTemplate.opsForValue().setIfAbsent(lockKey, lockVal, RedisConstant.ALBUM_LOCK_EXPIRE_PX2, TimeUnit.SECONDS);
            if (flag) {
                //3.获取锁成功执行业务,将查询业务数据放入缓存Redis
                log.info("获取锁成功：{}，线程名称：{}", Thread.currentThread().getId(), Thread.currentThread().getName());
                try {
                    albumInfo = this.getAlbumInfoFromDB(id);
                    long ttl = albumInfo == null ? RedisConstant.ALBUM_TEMPORARY_TIMEOUT : RedisConstant.ALBUM_TIMEOUT;
                    redisTemplate.opsForValue().set(dataKey, albumInfo, ttl, TimeUnit.SECONDS);
                    return albumInfo;
                } finally {
                    //4.业务执行完毕释放锁
                    String scriptText = "if redis.call(\"get\",KEYS[1]) == ARGV[1]\n" +
                            "then\n" +
                            "    return redis.call(\"del\",KEYS[1])\n" +
                            "else\n" +
                            "    return 0\n" +
                            "end";
                    DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>();
                    redisScript.setScriptText(scriptText);
                    redisScript.setResultType(Long.class);
                    redisTemplate.execute(redisScript, Arrays.asList(lockKey), lockVal);
                }
            } else {
                try {
                    //5.获取锁失败则自旋（业务要求必须执行）
                    Thread.sleep(200);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
                log.error("获取锁失败，自旋：{}，线程名称：{}", Thread.currentThread().getId(), Thread.currentThread().getName());
                return this.getAlbumInfo(id);
            }
        } catch (Exception e) {
            //兜底处理方案：Redis服务有问题，将业务数据获取自动从数据库获取
            log.info("[专辑服务]Redis服务异常：{}", e);
            return this.getAlbumInfoFromDB(id);
        }
    }*/


    /**
     * 根据专辑ID查询专辑信息（包含专辑标签及值）-从数据库获取
     *
     * @param id
     * @return
     */
    @Override
    @GuiGuCache(prefix="album:info:") //存入redis中业务数据key=album:info:1 锁key=album:info:1:lock
    public AlbumInfo getAlbumInfoFromDB(Long id) {
        //1.根据专辑ID主键查询专辑信息
        AlbumInfo albumInfo = albumInfoMapper.selectById(id);
        //2.根据专辑ID查询标签列表
        if (albumInfo != null) {
            LambdaQueryWrapper<AlbumAttributeValue> queryWrapper = new LambdaQueryWrapper<>();
            queryWrapper.eq(AlbumAttributeValue::getAlbumId, id);
            List<AlbumAttributeValue> attributeValueList = albumAttributeValueMapper.selectList(queryWrapper);
            albumInfo.setAlbumAttributeValueVoList(attributeValueList);
        }
        return albumInfo;
    }


    /**
     * 修改专辑信息
     *
     * @param id          专辑ID
     * @param albumInfoVo 修改专辑信息VO对象
     * @return
     */
    @Override
    @Transactional(rollbackFor = Exception.class)
    public void updateAlbumInfo(Long id, AlbumInfoVo albumInfoVo) {
        //1.修改专辑信息
        //2.1 将修改专辑信息转为PO对象 封装专辑ID
        AlbumInfo albumInfo = BeanUtil.copyProperties(albumInfoVo, AlbumInfo.class);
        albumInfo.setId(id);
        //2.2 修改专辑
        albumInfoMapper.updateById(albumInfo);

        //2.修改专辑标签 先删再新增
        //2.1 根据专辑ID将以前关联标签记录全部删除
        LambdaQueryWrapper<AlbumAttributeValue> queryWrapper = new LambdaQueryWrapper<>();
        queryWrapper.eq(AlbumAttributeValue::getAlbumId, id);
        albumAttributeValueMapper.delete(queryWrapper);

        //2.2 在根据提交专辑标签做新增
        List<AlbumAttributeValueVo> albumAttributeValueVoList = albumInfoVo.getAlbumAttributeValueVoList();
        if (CollectionUtil.isNotEmpty(albumAttributeValueVoList)) {
            //2.2 关联专辑ID批量保存专辑标签关系
            for (AlbumAttributeValueVo albumAttributeValueVo : albumAttributeValueVoList) {
                AlbumAttributeValue albumAttributeValue = BeanUtil.copyProperties(albumAttributeValueVo, AlbumAttributeValue.class);
                albumAttributeValue.setAlbumId(id);
                albumAttributeValueMapper.insert(albumAttributeValue);
            }
        }

        //TODO 专辑修改要先对专辑下架，修改，修改后审核通过后上架
        if ("1".equals(albumInfo.getIsOpen())) {
            kafkaService.sendMessage(KafkaConstant.QUEUE_ALBUM_UPPER, id.toString());
        } else {
            kafkaService.sendMessage(KafkaConstant.QUEUE_ALBUM_LOWER, id.toString());
        }
    }

    /**
     * 查询当指定用户专辑列表
     *
     * @param userId
     * @return
     */
    @Override
    public List<AlbumInfo> getUserAllAlbumList(Long userId) {
        //1.创建条件构造器封装查询条件
        LambdaQueryWrapper<AlbumInfo> queryWrapper = new LambdaQueryWrapper<>();
        queryWrapper.eq(AlbumInfo::getUserId, userId);
        //2.指定查询字段 select 后面列名称
        queryWrapper.select(AlbumInfo::getId, AlbumInfo::getAlbumTitle, AlbumInfo::getStatus);
        //3.TODO 避免小程序响应1千多条记录导致卡顿 需要指定加载数量  limit 200
        queryWrapper.last("limit 200");
        //4.设置排序按照最新专辑
        queryWrapper.orderByDesc(AlbumInfo::getId);
        return albumInfoMapper.selectList(queryWrapper);
    }

    /**
     * 根据专辑ID查询专辑统计信息
     *
     * @param albumId
     * @return
     */
    @Override
    @GuiGuCache(prefix="album:stat:")
    public AlbumStatVo getAlbumStatVo(Long albumId) {
        return albumInfoMapper.getAlbumStatVo(albumId);
    }

    /**
     * 分页获取专辑下声音列表，动态根据用户情况展示声音付费标识
     *
     * @param userId   用户ID
     * @param albumId  专辑ID
     * @param pageInfo 分页对象
     * @return
     */
    @Override
    public Page<AlbumTrackListVo> getAlbumTrackPage(Long userId, Long albumId, Page<AlbumTrackListVo> pageInfo) {
        //1.根据专辑ID分页获取该专辑下包含声音列表（包含声音统计信息）-默认声音付费标识为false
        pageInfo = albumInfoMapper.getAlbumTrackPage(pageInfo, albumId);
        //2.TODO 动态判断当前页中每个声音付费标识 关键点：找出付费情况
        //2.根据专辑ID查询专辑信息
        AlbumInfo albumInfo = albumInfoMapper.selectById(albumId);
        Assert.notNull(albumInfo, "专辑：{}不存在", albumId);
        String payType = albumInfo.getPayType();

        //3.处理用户未登录情况
        if (userId == null) {
            //3.1 判断专辑付费类型：VIP免费（0102）或 付费（0103） 除了免费试听外声音都应该设置付费标识
            if (SystemConstant.ALBUM_PAY_TYPE_VIPFREE.equals(payType) || SystemConstant.ALBUM_PAY_TYPE_REQUIRE.equals(payType)) {
                //3.2 获取本页中声音列表，过滤将声音序号大于免费试听集数声音付费标识设置为true
                pageInfo.getRecords()
                        .stream()
                        .filter(albumTrackVo -> albumTrackVo.getOrderNum() > albumInfo.getTracksForFree()) //过滤获取除免费试听以外声音
                        .collect(Collectors.toList())
                        .stream().forEach(albumTrackListVo -> {
                            albumTrackListVo.setIsShowPaidMark(true);
                        });
            }
        } else {
            //4.处理用户已登录情况
            //4.1 远程调用用户服务获取用户信息得到用户身份
            UserInfoVo userInfoVo = userFeignClient.getUserInfoVo(userId).getData();
            Assert.notNull(userInfoVo, "用户{}不存在", userId);
            Integer isVip = userInfoVo.getIsVip();

            //4.2 默认设置需要进一步确定购买情况标识：默认false
            Boolean isNeedCheckPayStatus = false;

            //4.2.1 如果专辑付费类型 VIP免费
            if (SystemConstant.ALBUM_PAY_TYPE_VIPFREE.equals(payType)) {
                //当前用户为普通用户或VIP会员过期
                if (isVip.intValue() == 0) {
                    isNeedCheckPayStatus = true;
                }
                if (isVip.intValue() == 1 && new Date().after(userInfoVo.getVipExpireTime())) {
                    isNeedCheckPayStatus = true;
                }
            }
            //4.2.2 如果专辑付费类型 付费
            if (SystemConstant.ALBUM_PAY_TYPE_REQUIRE.equals(payType)) {
                //当前用户为普通用户或VIP会员过期
                isNeedCheckPayStatus = true;
            }
            if (isNeedCheckPayStatus) {
                //4.3 进一步确定用户是否购买专辑或声音-远程调用用户服务获取本页中专辑或者声音购买情况
                //本页中需要检查购买情况声音列表，过滤掉当前页免费试听声音
                List<AlbumTrackListVo> needCheckTrackList = pageInfo.getRecords().stream()
                        .filter(albumTrackListVo -> albumTrackListVo.getOrderNum() > albumInfo.getTracksForFree())
                        .collect(Collectors.toList());

                //本页中需要检查购买情况声音ID列表，过滤掉当前页免费试听声音
                List<Long> needCheckTrackIdList = needCheckTrackList.stream()
                        .map(albumTrackListVo -> albumTrackListVo.getTrackId())
                        .collect(Collectors.toList());
                Map<Long, Integer> userPayStatusTrackMap = userFeignClient.userIsPaidTrack(userId, albumId, needCheckTrackIdList).getData();

                //4.4 循环当前页中声音列表-跟返回用户购买情况声音集合逐一判断
                needCheckTrackList.stream().forEach(needCheckTrack -> {
                    Integer payStatus = userPayStatusTrackMap.get(needCheckTrack.getTrackId());
                    if (payStatus.intValue() == 0) {
                        //4.5 某个声音用户未购买，将设置付费标识 isShowPaidMark:true
                        needCheckTrack.setIsShowPaidMark(true);
                    }
                });
            }
        }
        return pageInfo;
    }

    @Autowired
    private RedisTemplate redisTemplate;

    /**
     * MQ监听更新声音统计信息（包含：播放、收藏、点赞、评论）
     *
     * @param mqVo
     */
    @Override
    @Transactional(rollbackFor = Exception.class)
    public void updateTrackStat(TrackStatMqVo mqVo) {
        //1.做幂等性处理，统一个消息只处理一次 采用set k(业务消息唯一标识) v NX EX
        String key = "mq:" + mqVo.getBusinessNo();
        try {
            Boolean flag = redisTemplate.opsForValue().setIfAbsent(key, mqVo.getBusinessNo(), 1, TimeUnit.HOURS);
            if (flag) {
                //2.更新声音统计信息
                trackStatMapper.updateStat(mqVo.getTrackId(), mqVo.getStatType(), mqVo.getCount());

                //3.更新专辑统计信息(播放量、评论量只要声音+1，对应专辑也得+1)
                if (SystemConstant.TRACK_STAT_PLAY.equals(mqVo.getStatType())) {
                    albumStatMapper.updateStat(mqVo.getAlbumId(), SystemConstant.ALBUM_STAT_PLAY, mqVo.getCount());
                }
                if (SystemConstant.TRACK_STAT_COMMENT.equals(mqVo.getStatType())) {
                    albumStatMapper.updateStat(mqVo.getAlbumId(), SystemConstant.ALBUM_STAT_COMMENT, mqVo.getCount());
                }
            }
        } catch (Exception e) {
            //如果更新数据库发送异常，事务会进行回滚，下次再次投递消息允许继续处理统一个消息
            redisTemplate.delete(key);
            throw new RuntimeException(e);
        }
    }
}
